package com.lagou.es.controller;

import com.google.gson.Gson;
import com.lagou.domain.Products;
import com.lagou.vo.ProductsVO;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

@RestController
@RequestMapping("es")
public class EsController {

    @Autowired
    private KafkaConsumer kafkaConsumer;

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private Gson gson;

    /**
     * ES微服务将通过id查询到的商品对象，添加到Elasticsearch的lagou索引库中
     * @throws IOException
     */
    @GetMapping("/handleProductsMessage")
    public void handleProductsMessage() throws IOException {
        //获取product的消息
        kafkaConsumer.subscribe(Collections.singletonList("product"));
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) {
                String productsId = record.value();
                System.out.println("获取到的商品id：" + productsId);
                if (productsId != null) {
                    Products products = restTemplate.getForObject("http://localhost:9000/products/getProductsById/" + productsId, Products.class);
                    System.out.println(products);
                    //es添加索引
                    String source = gson.toJson(products);
                    //创建索引请求对象
                    IndexRequest request = new IndexRequest("lagou","product",products.getId()+"");
                    request.source(source, XContentType.JSON);
                    //发起请求
                    IndexResponse response = restHighLevelClient.index(request, RequestOptions.DEFAULT);
                    System.out.println("响应信息：" + response);
                }
            }
        }
    }

    /**
     *  实现商品查询功能，类似电商站内商品搜索
     *
     * 查询条件：name
     *
     * 分页查询：每页5条
     *
     * 排序规则：price，升序
     *
     * 结果过滤：商品价格范围过滤
     * @param productsVO
     * @return
     * @throws IOException
     */
    @PostMapping("/getProductsByName")
    public List<Products> getProductsByName(@RequestBody ProductsVO productsVO) throws IOException {
        SearchSourceBuilder builder = new SearchSourceBuilder();

        HighlightBuilder highlightBuilder = new HighlightBuilder();

        //高亮
        highlightBuilder.field("name");
        highlightBuilder.preTags("<font style='color:red'>").postTags("</font>");
        builder.highlighter(highlightBuilder);
        builder.query(QueryBuilders.matchQuery("name",productsVO.getName()));
        //添加排序
        builder.sort("price", SortOrder.ASC);
        //分页
        Integer pageNum = productsVO.getPageNum();
        Integer pageSize = productsVO.getPageSize();
        Integer start = (pageNum-1) * pageSize;
        builder.from(start);
        builder.size(pageSize);

        //调用封装的查询方法
        List<Products> productsList = basicQuery(builder);
        //对查询出的结果进行筛选  金额范围
        List<Products> list = productsList.stream().
                filter(products -> products.getPrice() > productsVO.getFrom() && products.getPrice() <= productsVO.getTo()).
                collect(Collectors.toList());
        return list;
    }

    /**
     * 封装的查询方法
     * @param sourceBuilder
     * @return
     * @throws IOException
     */
    private List<Products> basicQuery(SearchSourceBuilder sourceBuilder) throws IOException {
        List<Products> products = new ArrayList<>();
        //创建搜索请求对象
        SearchRequest request = new SearchRequest();
        //查询构建工具
        request.source(sourceBuilder);
        //执行查询
        SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
        //获得查询结果
        SearchHits hits = response.getHits();
        //获得文件数组
        SearchHit[] hitsHits = hits.getHits();
        for (SearchHit searchHit : hitsHits) {
            String hightlightName = "";
            //存放高亮的数据
            Map<String, HighlightField> map = searchHit.getHighlightFields();
            if (map != null) {
                Set<String> keySet = map.keySet();
                for (String s : keySet) {
                    HighlightField field = map.get(s);
                    Text[] texts = field.fragments();
                    for (Text text : texts) {
                        //高亮的 数据
                        hightlightName += text;
                    }
                }
            }
            String json = searchHit.getSourceAsString();
            //将json反序列化为Products格式
            Products products1 = gson.fromJson(json, Products.class);
            //设置到对应的需要高亮的字段中
            if (map != null) {
                products1.setName(hightlightName);
            }
            products.add(products1);
        }
        return products;
    }
}
